【精选】Spark写数据到Hadoop：HDFS与 Spark如何协同工作

Spark作为大数据计算框架，主要的优势在于数据计算，但是因为自身不具备分布式文件系统，所以往往需要结合其他的文件系统和数据源来完成工作，这其中就包括HDFS。HDFS是Hadoop的分布式文件系统，Spark写数据到Hadoop，其实就是HDFS与Spark协同工作，那么这一过程要如何实现呢？ Spark框架不管是写数据还是读取数据，基于Hadoop集群，主要靠的就是HDFS。我们可以把这个过程分为四个阶段： 将RDD写入HDFS→读取HDFS上的文件→将HDFS上的文件添加到Driver→判断HDFS上文件路径是否存在。  

Spark写数据到Hadoop，其实就是将RDD写入HDFS，当面对数据处理需求，向HDFS请求数据，读取RDD，然后才能进行后续的数据处理环节。第一步：启动Hadoop 首先启动Hadoop，在hadoop的目录下执行下面的命令： rm -rf tmp mkdir tmp cd sbin hadoop namenode -format start-dfs.sh start-yarn.sh 查看是否启动成功：第二步：将RDD写入HDFS 先创建一个SparkSession： val spark = SparkSession       .builder()       .appName("Spark SQL basic example")       .enableHiveSupport()       .getOrCreate()     spark.sparkContext.setLogLevel("WARN") 将RDD写入HDFS使用的函数是saveAsTextFile： val modelNames = Array("FM","FFM","DEEPFM","NFM","DIN","DIEN") val modelNamesRdd = spark.sparkContext.parallelize(modelNames,1) modelNamesRdd.saveAsTextFile("hdfs://localhost:9000/user/root/modelNames") 接下来，查看一下是否保存成功： 可以看到RDD在HDFS上是分块存储的，由于我们只有一个分区，所以只有part-0000。假设我们存储一个包含两个分区的RDD： val modelNames3 = Array("FM","FFM","DEEPFM","NFM","DIN","DIEN") val modelNames3Rdd = spark.sparkContext.parallelize(modelNames3,2) modelNames3Rdd.saveAsTextFile("hdfs://localhost:9000/user/root/modelNames3") 再次查看，可以看到有part-00000和part-00001:第三步：读取HDFS上的文件 读取HDFS上的文件，使用textFile方法： val modelNames2 = spark.sparkContext.textFile("hdfs://localhost:9000/user/root/modelNames/part-00000") val modelNames4 = spark.sparkContext.textFile("hdfs://localhost:9000/user/root/modelNames3/") 读取时是否加最后的part-00000都是可以的，当只想读取某个part，则必须加上。第四步：将HDFS上的文件添加到Driver 有时候，我们并不想直接读取HDFS上的文件，而是想对应的文件添加到Driver上，然后使用java或者Scala的I／O方法进行读取，此时使用addFile和get方法来实现： val files = "hdfs://localhost:9000/user/root/modelNames/part-00000" spark.sparkContext.addFile(files) val path = SparkFiles.get("part-00000") println(path) 打印的路径十分奇怪（没有截取完全）： 然后有了path之后，就可以使用scala的I／O进行读取： val source = Source.fromFile(path) val lineIterator = source.getLines val lines =lineIterator.toArray println(lines.mkString(",")) 输出为： FM,FFM,DEEPFM,NFM,DIN,DIEN第五步：判断HDFS上文件路径是否存在 在读取HDFS地址或者将文件传输到Driver上的时候，首先需要判断文件是否存在。单机环境下，代码如下： val conf = spark.sparkContext.hadoopConfiguration val path = new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames/part-00000") val fs = path.getFileSystem(conf) //得hdfs文件系统中的路径信息 val modelNamesExists = fs.exists(path) val modelNames1Exists = fs.exists(new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames1/part-00000")) println(modelNamesExists) println(modelNames1Exists) 输出结果为： true false 以上就是对Spark写数据到Hadoop这个过程的详细介绍了，当然这是基于单机环境来操作的，在真实的Hadoop集群环境当中，面临的数据以及细节可能会更复杂，但是基本流程就是这样完成的。